package com.fanxuankai.boot.monitor.confluent;

import com.alibaba.fastjson.JSON;
import com.fanxuankai.boot.monitor.Message;
import com.fanxuankai.boot.monitor.Monitor;
import com.fanxuankai.boot.monitor.Notifier;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @author fanxuankai
 */
public class ConfluentConsumerLagMonitor implements Monitor {
    @Resource
    private ConfluentConsumerLagProperties properties;
    @Resource
    private RestTemplate restTemplate;
    @Resource
    private Notifier<Message> notifier;

    @Override
    public String title() {
        return properties.getTitle();
    }

    @Override
    public String cron() {
        return properties.getCron();
    }

    @Override
    public void exec() {
        for (ConfluentConsumerLagProperties.ConsumerLagConfig config : properties.getConfigs()) {
            String healthStatusJson = restTemplate.getForObject(config.getClusters() + config.getHealthStatusUrl(),
                    String.class);
            String bootstrapClusterId = JSON.parseObject(healthStatusJson).getString("bootstrapClusterId");
            String url = config.getClusters() + config.getUrl() + "/" + bootstrapClusterId;
            String consumerGroupJson = restTemplate.getForObject(url, String.class);
            List<String> contents = Objects.requireNonNull(JSON.parseArray(consumerGroupJson, ConsumerGroupModel.class))
                    .stream()
                    .filter(o -> !StringUtils.isEmpty(o.getConsumerGroupId()))
                    .filter(o -> !o.getConsumerGroupId().startsWith("_confluent-controlcenter"))
                    .filter(o -> o.getTotalLag() > config.getThreshold())
                    .sorted((o1, o2) -> o2.getTotalLag() - o1.getTotalLag())
                    .map(o -> String.format("%s: %s", o.getConsumerGroupId(), o.getTotalLag()))
                    .collect(Collectors.toList());
            if (!CollectionUtils.isEmpty(contents)) {
                Message message = new Message();
                message.setTitle(properties.getTitle());
                message.setSubtitle(Optional.ofNullable(properties.getSubtitle())
                        .orElse(config.getClusters()));
                message.setContents(contents);
                message.setDescription("共计 " + contents.size() + " 个消费者组 Lag 过高");
                notifier.notice(message);
            }
        }
    }

    private static class ConsumerGroupModel {
        private String consumerGroupId;
        private long sumCurrentOffset;
        private long sumEndOffset;
        private int totalLag;
        private int numConsumers;
        private int numTopics;

        public String getConsumerGroupId() {
            return consumerGroupId;
        }

        public void setConsumerGroupId(String consumerGroupId) {
            this.consumerGroupId = consumerGroupId;
        }

        public long getSumCurrentOffset() {
            return sumCurrentOffset;
        }

        public void setSumCurrentOffset(long sumCurrentOffset) {
            this.sumCurrentOffset = sumCurrentOffset;
        }

        public long getSumEndOffset() {
            return sumEndOffset;
        }

        public void setSumEndOffset(long sumEndOffset) {
            this.sumEndOffset = sumEndOffset;
        }

        public int getTotalLag() {
            return totalLag;
        }

        public void setTotalLag(int totalLag) {
            this.totalLag = totalLag;
        }

        public int getNumConsumers() {
            return numConsumers;
        }

        public void setNumConsumers(int numConsumers) {
            this.numConsumers = numConsumers;
        }

        public int getNumTopics() {
            return numTopics;
        }

        public void setNumTopics(int numTopics) {
            this.numTopics = numTopics;
        }
    }
}
